package org.apache.hadoop.hive.cassandra.cql;

import java.util.Map;
import java.util.Properties;
import java.util.Set;

import org.apache.cassandra.thrift.ColumnDef;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.cassandra.CassandraException;
import org.apache.hadoop.hive.cassandra.input.cql.HiveCqlInputFormat;
import org.apache.hadoop.hive.cassandra.output.cql.HiveCqlOutputFormat;
import org.apache.hadoop.hive.cassandra.serde.AbstractCassandraSerDe;
import org.apache.hadoop.hive.cassandra.serde.cql.CqlSerDe;
import org.apache.hadoop.hive.metastore.HiveMetaHook;
import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
import org.apache.hadoop.hive.ql.metadata.DefaultStorageHandler;
import org.apache.hadoop.hive.ql.plan.TableDesc;
import org.apache.hadoop.hive.serde2.SerDe;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapred.OutputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public class Cql2StorageHandler extends DefaultStorageHandler {
	
	private static final Logger logger = LoggerFactory.getLogger(Cql2StorageHandler.class);

	private static Configuration configuration;

	@Override
	public void setConf(Configuration configuration) {
		this.configuration = configuration;
	}

	@Override
	public Configuration getConf() {
		return this.configuration;
	}

	@Override
	public Class<? extends InputFormat> getInputFormatClass() {
		return HiveCqlInputFormat.class;
	}

	@Override
	public Class<? extends OutputFormat> getOutputFormatClass() {
		return HiveCqlOutputFormat.class;
	}

	@Override
	public Class<? extends SerDe> getSerDeClass() {
		return CqlSerDe.class;
	}
	
	@Override
	  public void configureTableJobProperties(final TableDesc tableDesc, final Map<String, String> jobProperties) {
	    configureJob(tableDesc.getProperties(), jobProperties);
	  }
	public static void configureJob(final Properties tableProperties, final Map<String, String> jobProperties){
		//Identify Keyspace
        String keyspace = tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_KEYSPACE_NAME);
        if (keyspace == null) {
            keyspace = tableProperties.getProperty(hive_metastoreConstants.META_TABLE_DB);
        }

        jobProperties.put(AbstractCassandraSerDe.CASSANDRA_KEYSPACE_NAME, keyspace);

        //Identify ColumnFamily
        String columnFamily = tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_CF_NAME);
        if (columnFamily == null) {
            columnFamily = tableProperties.getProperty(hive_metastoreConstants.META_TABLE_NAME);
        }

        jobProperties.put(AbstractCassandraSerDe.CASSANDRA_CF_NAME, columnFamily);

        //If no column mapping has been configured, we should create the default column mapping.
        String columnInfo = tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_COL_MAPPING);
        if (columnInfo == null) {
            columnInfo = CqlSerDe.createColumnMappingString(
                    tableProperties.getProperty(org.apache.hadoop.hive.serde.Constants.LIST_COLUMNS));
        }
        jobProperties.put(AbstractCassandraSerDe.CASSANDRA_COL_MAPPING, columnInfo);

        String host = configuration.get(AbstractCassandraSerDe.CASSANDRA_HOST);
        if (host == null) {
            host = tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_HOST, AbstractCassandraSerDe.DEFAULT_CASSANDRA_HOST);
        }

        jobProperties.put(AbstractCassandraSerDe.CASSANDRA_HOST, host);

        String port = configuration.get(AbstractCassandraSerDe.CASSANDRA_PORT);
        if (port == null) {
            port = tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_PORT, AbstractCassandraSerDe.DEFAULT_CASSANDRA_PORT);
        }

        jobProperties.put(AbstractCassandraSerDe.CASSANDRA_PORT, port);

        String maxThreads = configuration.get(AbstractCassandraSerDe.CASSANDRA_HADOOP_MAX_THREADS_KEY);
        if (maxThreads == null) {
            maxThreads = tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_HADOOP_MAX_THREADS_KEY, AbstractCassandraSerDe.DEFAULT_CASSANDRA_HADOOP_MAX_THREADS);
        }
        jobProperties.put(AbstractCassandraSerDe.CASSANDRA_HADOOP_MAX_THREADS_KEY, maxThreads);

        String maxRetries = configuration.get(AbstractCassandraSerDe.CASSANDRA_HADOOP_RETRIES_KEY);
        if (maxRetries == null) {
            maxRetries = tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_HADOOP_RETRIES_KEY, AbstractCassandraSerDe.DEFAULT_CASSANDRA_HADOOP_RETRIES);
        }
        jobProperties.put(AbstractCassandraSerDe.CASSANDRA_HADOOP_RETRIES_KEY, maxRetries);


        if (configuration.get(AbstractCassandraSerDe.CASSANDRA_PARTITIONER) == null) {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_PARTITIONER,
                    tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_PARTITIONER,
                            "org.apache.cassandra.dht.Murmur3Partitioner"));
        } else {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_PARTITIONER, configuration.get(AbstractCassandraSerDe.CASSANDRA_PARTITIONER));
        }

        if (configuration.get(AbstractCassandraSerDe.CASSANDRA_CONSISTENCY_LEVEL) == null) {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_CONSISTENCY_LEVEL,
                    tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_CONSISTENCY_LEVEL,
                            AbstractCassandraSerDe.DEFAULT_CONSISTENCY_LEVEL));
        } else {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_CONSISTENCY_LEVEL, configuration.get(AbstractCassandraSerDe.CASSANDRA_CONSISTENCY_LEVEL));
        }

        if (configuration.get(AbstractCassandraSerDe.CASSANDRA_RANGE_BATCH_SIZE) == null) {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_RANGE_BATCH_SIZE,
                    tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_RANGE_BATCH_SIZE,
                            Integer.toString(AbstractCassandraSerDe.DEFAULT_RANGE_BATCH_SIZE)));
        } else {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_RANGE_BATCH_SIZE, configuration.get(AbstractCassandraSerDe.CASSANDRA_RANGE_BATCH_SIZE));
        }

        if (configuration.get(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_SIZE) == null) {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_SIZE,
                    tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_SIZE,
                            Integer.toString(AbstractCassandraSerDe.DEFAULT_SLICE_PREDICATE_SIZE)));
        } else {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_SIZE, configuration.get(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_SIZE));
        }

        if (configuration.get(AbstractCassandraSerDe.CASSANDRA_SPLIT_SIZE) == null) {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SPLIT_SIZE,
                    tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_SPLIT_SIZE,
                            Integer.toString(AbstractCassandraSerDe.DEFAULT_SPLIT_SIZE)));
        } else {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SPLIT_SIZE, configuration.get(AbstractCassandraSerDe.CASSANDRA_SPLIT_SIZE));
        }

        if (configuration.get(AbstractCassandraSerDe.CASSANDRA_BATCH_MUTATION_SIZE) == null) {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_BATCH_MUTATION_SIZE,
                    tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_BATCH_MUTATION_SIZE,
                            Integer.toString(AbstractCassandraSerDe.DEFAULT_BATCH_MUTATION_SIZE)));
        } else {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_BATCH_MUTATION_SIZE, configuration.get(AbstractCassandraSerDe.CASSANDRA_BATCH_MUTATION_SIZE));
        }

        if (configuration.get(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_START) == null) {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_START,
                    tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_START, ""));
        } else {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_START, configuration.get(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_START));
        }

        if (configuration.get(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_FINISH) == null) {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_FINISH,
                    tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_FINISH, ""));
        } else {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_FINISH, configuration.get(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_FINISH));
        }

        if (configuration.get(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_COMPARATOR) == null) {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_COMPARATOR,
                    tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_COMPARATOR, ""));
        } else {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_COMPARATOR,
                    configuration.get(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_COMPARATOR));
        }

        if (configuration.get(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_REVERSED) == null) {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_REVERSED,
                    tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_REVERSED, "false"));
        } else {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_REVERSED,
                    configuration.get(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_REVERSED));
        }

        //Set the indexed column names - leave unset if we have problems determining them
        String indexedColumns = tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_INDEXED_COLUMNS);
        if (indexedColumns != null) {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_INDEXED_COLUMNS, indexedColumns);
        } else {
            try {
                Set<ColumnDef> columns = CqlPushdownPredicate.getIndexedColumns(host, Integer.parseInt(port), keyspace, columnFamily);
                jobProperties.put(AbstractCassandraSerDe.CASSANDRA_INDEXED_COLUMNS, CqlPushdownPredicate.serializeIndexedColumns(columns));
            } catch (CassandraException e) {
                // this results in the property remaining unset on the Jobconf, so indexes will not be used on the C* side
                logger.info("Error determining cassandra indexed columns, will not include in JobConf", e);
            }
        }
	}
	

	  @Override
	  public HiveMetaHook getMetaHook() {
	    return null;
	  }
}
